package com.netty.connect;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.netty.handler.RpcClientHandler;
import com.netty.handler.RpcClientInitializer;

/**
 * 连接管理
 * @author dwjf321
 *
 */
public class ConnectManage {
	private static final Logger logger = LoggerFactory.getLogger(ConnectManage.class);
	private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);
	private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16,
            600L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));
	
	private CopyOnWriteArrayList<RpcClientHandler> connectedHandlers = new CopyOnWriteArrayList<RpcClientHandler>();
	private Map<InetSocketAddress, RpcClientHandler> connectedServerNodes = new ConcurrentHashMap<InetSocketAddress, RpcClientHandler>();
	
	private ReentrantLock lock = new ReentrantLock();
	private Condition connected = lock.newCondition();
	private long connectTimeoutMillis = 6000;
	private AtomicInteger roundRobin = new AtomicInteger(0);
	private volatile boolean isRuning = true;
	
	private volatile static ConnectManage connectManage;
	private ConnectManage(){}
	public static ConnectManage getInstance(){
		if (connectManage == null) {
			synchronized (ConnectManage.class) {
				if (connectManage == null) {
					connectManage = new ConnectManage();
				}
			}
		}
		return connectManage;
	}
	
	
	public void updateConnectedServer(List<String> allServerAddress) {
		if (allServerAddress != null) {
			if (allServerAddress.size() > 0) {//得到有效的服务节点
				//更新本地服务节点缓存
				HashSet<InetSocketAddress> newAllServerNodeSet = new HashSet<InetSocketAddress>();
				for (int i = 0; i < allServerAddress.size(); i++) {
					String[] array = allServerAddress.get(i).split(":");
					if (array.length == 2) {
						String host = array[0];
						int port = Integer.parseInt(array[1]);
						final InetSocketAddress remotePeer = new InetSocketAddress(host, port);
						newAllServerNodeSet.add(remotePeer);
					}
				}
				//添加新的服务节点
				for (final InetSocketAddress serverNodeAddress : newAllServerNodeSet) {
					if (!connectedServerNodes.keySet().contains(serverNodeAddress)) {
						connectServerNode(serverNodeAddress);
					}
				}
			}else {//没有可用服务，所有的服务宕机
				logger.error("没有可用服务，所有的服务宕机");
				for (final RpcClientHandler connectedServerHandler : connectedHandlers) {
					SocketAddress remotePeer = connectedServerHandler.getRemotePeer();
					RpcClientHandler handler = connectedServerNodes.get(remotePeer);
					handler.close();
					connectedServerNodes.remove(connectedServerHandler);
				}
				connectedHandlers.clear();
				
			}
		}
		
	}
	/**
	 * 重连
	 * @param handler
	 * @param remotePeer
	 */
	public void reconnect(final RpcClientHandler handler, final SocketAddress remotePeer){
		if (handler != null) {
			connectedHandlers.remove(handler);
			connectedServerNodes.remove(handler.getRemotePeer());
		}
		connectServerNode((InetSocketAddress)remotePeer);
	}
	
	/**
	 * 连接服务器节点
	 * @param serverNodeAddress
	 */
	private void connectServerNode(final InetSocketAddress remotePeer) {
		threadPoolExecutor.submit(new Runnable() {
			
			@Override
			public void run() {
				Bootstrap bootstrap = new Bootstrap();
				bootstrap.group(eventLoopGroup)
					.channel(NioSocketChannel.class)
					.handler(new RpcClientInitializer());
				ChannelFuture channelFuture = bootstrap.connect(remotePeer);
				channelFuture.addListener(new ChannelFutureListener() {
					
					@Override
					public void operationComplete(ChannelFuture future) throws Exception {
						if (channelFuture.isSuccess()) {
							logger.debug("成功连接服务器："+remotePeer);
							RpcClientHandler handler = channelFuture.channel().pipeline().get(RpcClientHandler.class);
							addHandler(handler);
						}
					}
				});
			}
		});
		
	}
	/**
	 * 添加处理handler
	 * @param handler
	 */
	private void addHandler(RpcClientHandler handler) {
		connectedHandlers.add(handler);
		InetSocketAddress remoteAddress = (InetSocketAddress) handler.getChannel().remoteAddress();
		connectedServerNodes.put(remoteAddress, handler);
		signalAvailableHandler();
	}
	/**
	 * 通知所有的处理器
	 */
	private void signalAvailableHandler() {
		lock.lock();
		try {
			connected.signalAll();
		} finally{
			lock.unlock();
		}
		
	}
	/**
	 * 等待时间
	 */
	private boolean waitingForHandler() throws InterruptedException {
		lock.lock();
		try{
			return connected.await(this.connectTimeoutMillis, TimeUnit.MILLISECONDS);
		}finally{
			lock.unlock();
		}
	}
	/**
	 * 选择处理器对象
	 * @return
	 */
	public RpcClientHandler chooseHandler(){
		int size = connectedHandlers.size();
		while (isRuning && size <= 0) {
			try {
				boolean available = waitingForHandler();
				if (available) {
					size = connectedHandlers.size();
				}
			} catch (InterruptedException e) {
				logger.error("等待有效的节点中断",e);
				throw new RuntimeException("不能连接任何服务",e);
			}
		}
		int index = (roundRobin.getAndAdd(1)+size)%size;
		return connectedHandlers.get(index);
	}
	
	/**
	 * 停止
	 */
	public void stop() {
		isRuning = false;
		for (int i = 0; i < connectedHandlers.size(); i++) {
			RpcClientHandler connectedServerHandler = connectedHandlers.get(i);
			connectedServerHandler.close();
		}
		signalAvailableHandler();
		threadPoolExecutor.shutdown();
		eventLoopGroup.shutdownGracefully();
	}
}
